//
// Created by suzhen on 17-5-3.
//

#ifndef ORIONV2_SHAREDPROMISE_H
#define ORIONV2_SHAREDPROMISE_H

#include "Orion.h"
#include "Traits.hpp"

ORION_V2_NAMESPACE_BEGIN

        template<typename T>
        using thread_future = boost::unique_future<T>;

        template<typename T>
        using thread_shared_future = boost::shared_future<T>;

        template<typename T>
        using thread_promise = boost::promise<T>;

        template<typename T>
        using fiber_future = boost::fibers::future<T>;

        template<typename T>
        using fiber_shared_future = boost::fibers::shared_future<T>;

        template<typename T>
        using fiber_promise = boost::fibers::promise<T>;

        typedef typename boost::fibers::recursive_mutex FiberRecursiveMutex;

        typedef typename std::recursive_mutex ThreadRecursiveMutex;

        template<typename T, typename T2>
        class SmartFuture;

        template<typename T, typename T2>
        class SmartPromise;

        template<typename ValueType,
                typename MutexType>
        class SmartPromiseCore {
            friend class SmartFuture<ValueType,
                    MutexType>;

            friend class SmartPromise<ValueType,
                    MutexType>;

        public:
            typedef std::function<void(const ValueType &)> CallBackType;

            SmartPromiseCore()
                    : internalPms() {
            }

            SmartPromiseCore(ValueType &&v)
                    : internalPms() {
                internalPms.set_value(std::forward<ValueType>(v));
            }

            void addListener(CallBackType &&fc) {
                std::lock_guard<MutexType> guard(mtx_);
                watcherList.emplace_back(std::forward<CallBackType>(fc));
            }

            void notify(ValueType &val) {
                std::lock_guard<MutexType> guard(mtx_);
                while (!watcherList.empty()) {
                    auto &fnt = watcherList.front();
                    fnt(val);
                    watcherList.pop_front();
                }
            }

            //todo 优化:这里的 recursive_mutex 可以使用 spin_lock
        private:
            MutexType mtx_;
            std::list<CallBackType> watcherList;
            thread_promise<ValueType> internalPms;

        };

        template<typename ValueType,
                typename MutexType
        >
        class SmartPromise {
        public:
            typedef ValueType value_type;
            typedef SmartPromiseCore<ValueType, MutexType> CoreType;
            using PromiseTp = thread_promise<ValueType>;
            using FutureTp = thread_shared_future<ValueType>;

        public:
            SmartPromise()
                    : pmsCore_(std::make_shared<CoreType>()),
                      tsf_(pmsCore_->internalPms.get_future().share()) {
            }

            SmartPromise(ValueType &&v)
                    : pmsCore_(
                    std::make_shared<CoreType>(ValueType(std::forward<ValueType>(v)))),
                      tsf_(pmsCore_->internalPms.get_future().share()) {
            }

            SmartPromise(const SmartPromise &) = delete;

            SmartPromise &operator=(const SmartPromise &) = delete;

            SmartPromise(SmartPromise &&o)
                    : pmsCore_(std::move(o.pmsCore_)),
                      tsf_(std::move(o.tsf_)) {
            }

            ~SmartPromise() {
                if (pmsCore_ && !isReady()) {
                    SYS_FAULT(false) << "fatal error";
                }
            }

            SmartPromise &operator=(SmartPromise &&o) {
                if (this == &o) {
                    return *this;
                }
                pmsCore_ = std::move(o.pmsCore_);
                tsf_ = std::move(o.tsf_);
                return *this;
            }

            SmartFuture<ValueType, MutexType> getFuture() {
                return SmartFuture<ValueType, MutexType>(pmsCore_, tsf_);
            }

            bool isReady() {
                return tsf_.is_ready();
            }

            bool hasFullFilled() {
                return isReady() && hasValue();
            }

            bool hasValue() {
                return tsf_.has_value();
            }

            template<typename... Arg>
            void setValue(Arg &&... args) {
                SYS_FAULT(!isReady()) << "promise already fullfilled";
                SYS_FAULT(pmsCore_) << "broken promise";
                ValueType result(std::forward<Arg>(args)...);
                //notify all observers
                pmsCore_->notify(result);
                //set ok
                pmsCore_->internalPms.set_value(std::move(result));
            }

        private:
            std::shared_ptr<CoreType> pmsCore_;
            FutureTp tsf_;
        };

        /**
         * thread promise
         */
        template<typename T>
        using Promise = SmartPromise<T, ThreadRecursiveMutex>;

        /**
         * fiber promise
         */
        template<typename T>
        using FiberPromise = SmartPromise<T, FiberRecursiveMutex>;

        template<typename ValueType,
                typename MutexType>
        class SmartFuture {
        public:
            typedef SmartPromise<ValueType, MutexType> PromiseTp;
            typedef typename PromiseTp::FutureTp FutureRawType;
            typedef ValueType value_type;

            ~SmartFuture() {
            }

            SmartFuture(std::shared_ptr<typename PromiseTp::CoreType> core,
                        const FutureRawType &o)
                    : pmsCore_(core),
                      rawFuture_(o) {
            }

            template<typename F>
            void then__(F &&f,
                        std::true_type
            ) {
                if (rawFuture_.is_ready()) {
                    f(rawFuture_.get());
                    return;
                }
                std::lock_guard<MutexType> guard(pmsCore_->mtx_);
                if (pmsCore_) {
                    pmsCore_->addListener(std::forward<F>(f));
                }
            }

            template<typename F>
            void then__(F &&f,
                        std::false_type) {
                if (rawFuture_.is_ready()) {
                    f();
                    return;
                }
                then__([f = std::move(f)](const ValueType &) mutable {
                    f();
                }, std::true_type{});
            }

            template<typename F>
            void then(F &&f) {
                then__(std::forward<F>(f),
                       std::is_same<
                               typename TypeTraits::ExtractMemberFunction<decltype(&F::operator())>::FirstArg,
                               typename std::add_lvalue_reference_t<typename std::add_const_t<ValueType>>>{});
            }

            const ValueType &get() const {
                return rawFuture_.get();
            }

            template<typename _Rep, typename _Period>
            void waitValueFor(const boost::chrono::duration<_Rep, _Period> &d) const {
                rawFuture_.wait_for(d);
            }

            bool isReady() const {
                return rawFuture_.is_ready()
                       && rawFuture_.has_value();
            }

        private:
            std::shared_ptr<typename PromiseTp::CoreType> pmsCore_;
            FutureRawType rawFuture_;
        };

        template<typename T, typename T2>
        std::vector<T> collectAll(std::vector<SmartFuture<T, T2>> fv) {
            std::vector<T> ret;
            std::for_each(fv.begin(), fv.end(), [&](SmartFuture<T, T2> &f) {
                ret.emplace_back(f.get());
            });
            return ret;
        }

        template<typename T, typename M>
        SmartFuture<T, M> makeReadyFuture(T &&o) {
            SmartPromise<T, M> pms(std::forward<T>(o));
            return pms.getFuture();
        }

        template<typename T>
        SmartFuture<T, FiberRecursiveMutex> makeReadyFiberFuture(T &&o) {
            return makeReadyFuture<T, FiberRecursiveMutex>(std::forward<T>(o));
        }

        template<typename T>
        SmartFuture<T, ThreadRecursiveMutex> makeReadyThreadFuture(T &&o) {
            return makeReadyFuture<T, ThreadRecursiveMutex>(std::forward<T>(o));
        }

        /**
         * thread future
         */
        template<typename T>
        using Future = SmartFuture<T, ThreadRecursiveMutex>;

        /**
         * fiber future
         */
        template<typename T>
        using FiberFuture = SmartFuture<T, FiberRecursiveMutex>;

ORION_V2_NAMESPACE_END


#endif //ORIONV2_SHAREDPROMISE_H
